#ifndef MQSERVER_MQ_MESSAGE_HPP_
#define MQSERVER_MQ_MESSAGE_HPP_
#include "../mqcomm/mq_helper.hpp"
#include "../mqcomm/mq_logger.hpp"
#include "../mqcomm/proto/mq_msg.pb.h"
#include <algorithm>
#include <atomic>
#include <cassert>
#include <cstddef>
#include <iostream>
#include <list>
#include <memory>
#include <mutex>
#include <string>
#include <unordered_map>

namespace mqserver
{
class QueueMessageManager;
class MessageManager;
using MessagePtr = std::shared_ptr<proto::message::Message>;
using QueueMessageManagerPtr = std::shared_ptr<QueueMessageManager>;
using MessageManagerPtr = std::shared_ptr<MessageManager>;

class MessageMapper
{
	const std::string default_subifx = ".mqd";
	const std::string tmpfile_subifx = ".mqd.tmp";
public:
	MessageMapper() = default;
	MessageMapper(std::string* basedir, const std::string& qname) 
		: qname_(qname)
	{
		if (basedir->back() != '/') basedir->push_back('/');
		datafile_ = *basedir + qname + default_subifx;
		tmpfile_ = *basedir + qname + tmpfile_subifx;
		assert(helper::FileHelper::createDirectory(*basedir));
		createMsgFile();
	}
	bool createMsgFile()
	{	
		if(false == helper::FileHelper::createFile(datafile_))
		{
			LOG_FATAL << "create file: \"" << qname_ << "\" fail!\n";
			return false;
		}
		return true;
	}
	bool removeMsgFile()
	{
		return helper::FileHelper::removeFile(datafile_)
			&& helper::FileHelper::removeFile(tmpfile_);
	}
	bool insert(MessagePtr msg)
	{
		return insert(datafile_, msg);
	}
	bool remove(MessagePtr msg)
	{
		// 1. 修改标志位Payload.valid=0
		msg->mutable_payload()->set_valid("0");
		// 2. msg序列化
		std::string body = msg->payload().SerializePartialAsString();
		if(body.size() != msg->length()) 
		{
			LOG_ERROR << "The length of the serialized data \
							is inconsistent with the original record\n";
			return false;
		}
		// 3.覆盖重写
		helper::FileHelper file_helper(datafile_);
		if(false == file_helper.write(body.c_str(), msg->offset(), body.size()))
		{
			LOG_ERROR << "write msgqueue's body fail!\n";
			return false;
		}
		return true;
	}
	std::list<MessagePtr> garbageRecycle()
	{
		std::list<MessagePtr> result;
		// 1. 加载有效数据
		if(false == load(&result))
		{
			LOG_ERROR << "load valid data fail!\n";
			return result;
		}
		// 2.有效数据存储到临时文件tmpfile
		helper::FileHelper::createFile(tmpfile_);
		for(auto& msg : result)
		{
			if(insert(msg) == false) 
			{
				LOG_ERROR << "write data to tmp_file fail!\n";
				return result;
			}
		}
		// 3. 删除源文件datafile, 重命名tmpfile-->datafile
		if(false == helper::FileHelper::removeFile(datafile_))
		{
			LOG_ERROR << "remove srcfile fail!\n";
			return result;
		}
		if(false == helper::FileHelper(tmpfile_).rename(datafile_))
		{
			LOG_ERROR << "rename tmpfile fail!\n";
			return result;
		}
		// 4. 返回新的有效数据
		return result;
	}
	void clear() { removeMsgFile(); }
private:
	bool load(std::list<MessagePtr>* result)
	{
		// 读文件, len.data|len.data|...
		helper::FileHelper file_helper(datafile_);
		size_t fsize = file_helper.size();
		size_t offset = 0, msg_size = 0;
		while (offset < fsize)
		{
			if(false == file_helper.read((char*)&msg_size, offset, sizeof(msg_size))) 
			{
				LOG_ERROR << "read msgqueue's len fail!\n";
				return false;
			}
			offset += sizeof(msg_size);
			std::string msg_body(msg_size, '\0');
			if(false == file_helper.read(&msg_body[0], offset, msg_size))
			{
				LOG_ERROR << "read msgqueue's body fail!\n";
				return false;
			}
			offset += msg_size;
			MessagePtr msgp = std::make_shared<proto::message::Message>();
			msgp->mutable_payload()->ParseFromString(msg_body);				// 反序列化到Message中
			if(msgp->payload().valid() == "0") continue;	// 无效数据不添加
			result->push_back(msgp);
		}
		return true;
	}
	bool insert(const std::string& filename, MessagePtr msg)
	{
		// 1. 消息序列化
		std::string body = msg->payload().SerializeAsString();
		size_t msg_size = body.size();
		// 2. 获取文件大小, 从 fsize 开始往后写入新数据
		helper::FileHelper file_helper(filename);
		size_t offset = file_helper.size();
		// 2.1 先写数据长度
		if(false == file_helper.write((char*)&msg_size, offset, sizeof(msg_size)))
		{
			LOG_ERROR << "write msgqueue's len fail!\n";
			return false;
		}
		offset += sizeof(msg_size);
		// 2.2 写正文数据
		if(false == file_helper.write(body.c_str(), offset, body.size()))
		{
			LOG_ERROR << "write msgqueue's body fail!\n";
			return false;
		}
		// 3. 更新msg实际存储信息(偏移量/长度)
		msg->set_offset(offset);
		msg->set_length(body.size());
		return true;
	}
private:
	std::string qname_;
	std::string datafile_;
	std::string tmpfile_;
};

class QueueMessageManager
{
public:
	QueueMessageManager() = default;
	QueueMessageManager(std::string* basedir, const std::string& qname)
		: mapper_(basedir, qname), qname_(qname)
		, vaild_count_(0), total_count_(0) {}

	bool recovery()
	{
		std::lock_guard<std::mutex> lock(mtx_);
		waitpush_msgs_ = mapper_.garbageRecycle();
		for(auto& msg : waitpush_msgs_)
			durable_msgs_.emplace(msg->payload().properties().id(), msg);
		vaild_count_ = total_count_ = waitpush_msgs_.size();
		return true;
	}
	bool insert(const proto::message::BasicProperties* basic_props, 
				const std::string& body, bool queue_durable) 
	{
		// 1. 构造消息对象(局部数据, 没有线程安全问题)		
		MessagePtr message = ConstructMessage(basic_props, body, queue_durable);	
		// 2. 判断是否需要持久化
		std::lock_guard<std::mutex> lock(mtx_);
		if(message->payload().properties().delivery_mode() == proto::message::DeliveryMode::DURABLE)
		{
			message->mutable_payload()->set_valid("1");
			if(false == mapper_.insert(message)) 
			{
				LOG_ERROR << "message: \"" << body << "\" durable fail!\n";
				return false;
			}
			++vaild_count_;
			++total_count_;
			durable_msgs_.emplace(message->payload().properties().id(), message);
		}
		// 3. 内存管理
		waitpush_msgs_.push_back(message);
		return true;
	}
	MessagePtr front()
	{
		std::lock_guard<std::mutex> lock(mtx_);
		if(waitpush_msgs_.size() <= 0) return MessagePtr();
		// 返回之前, 需要先加入 waitack_msgs_
		MessagePtr message = waitpush_msgs_.front();
		waitpush_msgs_.pop_front();
		waitack_msgs_.emplace(message->payload().properties().id(), message);
		return message;
	}	
	bool remove(const std::string& msg_id) 	// 删除已经推送且为确认的消息
	{
		// 1. 查 waitack_msgs_
		auto it = waitack_msgs_.find(msg_id);
		if(it == waitack_msgs_.end())
		{
			LOG_WARNING << "msg_id: \"" << msg_id << "\" dones't exists!\n";
			return true;
		}
		// 2. 判断是否需要删持久化数据
		if(it->second->payload().properties().delivery_mode() == proto::message::DeliveryMode::DURABLE)
		{
			mapper_.remove(it->second);
			durable_msgs_.erase(msg_id);
			--vaild_count_;					// 持久化有效数量-1
			garbageRecycle();  				// 内部垃圾回收检测
		}
		// 3. 删内存
		waitack_msgs_.erase(msg_id);
		return true;
	}

	size_t vaildCount()
	{
		std::lock_guard<std::mutex> lock(mtx_);
		return vaild_count_;
	}
	size_t durableCount()
	{
		std::lock_guard<std::mutex> lock(mtx_);
		return durable_msgs_.size();
	}
	size_t totalCount()
	{
		std::lock_guard<std::mutex> lock(mtx_);
		return total_count_;
	}
	size_t pendingPushCount()
	{
		std::lock_guard<std::mutex> lock(mtx_);
		return waitpush_msgs_.size();
	}
	size_t pendingAckCount()
	{
		std::lock_guard<std::mutex> lock(mtx_);
		return waitack_msgs_.size();
	}
	void clear()
	{
		std::lock_guard<std::mutex> lock(mtx_);
		mapper_.clear();
		waitpush_msgs_.clear();
		waitack_msgs_.clear();
		durable_msgs_.clear();
		vaild_count_ = 0;
		total_count_ = 0;
	}
private:
	MessagePtr ConstructMessage(const proto::message::BasicProperties* basic_props, 
								const std::string& body, bool queue_durable) 
	{
		MessagePtr message = std::make_shared<proto::message::Message>();
		message->mutable_payload()->set_body(body);

		bool msg_durable = basic_props ?  	// bp存在则代表自己, 不存在则默认DURABLE	
			basic_props->delivery_mode() == proto::message::DeliveryMode::DURABLE : true;
		// 二者都为真, 代表msg为DURABLE或不存在, 那么msg都是选择DURABLE, msg&&queue就是DURABLE
		// 二者其一为假, 代表msg为UNDURABLE或queue为UNDURABLE, 那么都是选择UNDURABLE
		proto::message::DeliveryMode msg_mode = msg_durable && queue_durable ?			
			proto::message::DeliveryMode::DURABLE : proto::message::DeliveryMode::UNDURABLE;	

		std::string msg_id = basic_props ? basic_props->id() : helper::UUIDHelper::uuid();
		std::string msg_key = basic_props ? basic_props->routing_key() : "";	
		
		auto properties = message->mutable_payload()->mutable_properties();
		properties->set_id(msg_id);
		properties->set_delivery_mode(msg_mode);
		properties->set_routing_key(msg_key);

		return std::move(message);
	}
	bool garbageRecycleCheck()
	{
		return total_count_ > 2000 
			&& vaild_count_ * 10 / total_count_ < 5;
	}
	void garbageRecycle()
	{
		if(!garbageRecycleCheck()) return;
		// 1. 获取垃圾回收后的有效信息
		std::list<MessagePtr> new_msgs = mapper_.garbageRecycle();
		for(auto& msg : new_msgs) 
		{
			std::string msg_id = msg->payload().properties().id();
			auto it = durable_msgs_.find(msg_id);
			if(it == durable_msgs_.end())	// 理应不存在的情况
			{
				LOG_WARNING << "durable_msgId: \"" << msg_id
							<< "\" doesn't exists after garbagerecycle!\n";
				waitpush_msgs_.push_back(it->second);
				durable_msgs_.emplace(msg_id, it->second);
				continue;
			}
			// 2. 更新实际存储位置
			it->second->set_offset(msg->offset());
			it->second->set_length(msg->length());
		}
		// 3. 更新当前 vaild_count_ && total_count_
		vaild_count_ = total_count_ = new_msgs.size();
	}
private:
	std::mutex mtx_;
	std::string qname_;		// 队列名
	std::list<MessagePtr> waitpush_msgs_;		// 待推送消息
	std::unordered_map<std::string, MessagePtr> waitack_msgs_;  // 待确认消息<msg_id, msg>
	std::unordered_map<std::string, MessagePtr> durable_msgs_;	// 持久化消息<msg_id, msg>
	MessageMapper mapper_;	  // 持久化句柄
	size_t vaild_count_;	  // 持久化中的有效消息数量
	size_t total_count_;	  // 持久化中的消息总理
};

class MessageManager 
{
public:
	MessageManager(const std::string& basedir) : basedir_(basedir) {}

	void initMessageQueue(const std::string& qname)		// 初始化指定队列的消息管理句柄(恢复队列消息数据)
	{
		QueueMessageManagerPtr queue_msg;		// 从queue_msgs取出队列消息管理句柄, 减少加锁的时间
		{
			std::lock_guard<std::mutex> lock(mtx_);
			if(existsNoLock(qname))	return;				// 队列已存在
			queue_msg = std::make_shared<QueueMessageManager>(&basedir_, qname);
			queue_msgs.emplace(qname, queue_msg);
		}
		queue_msg->recovery();		// 涉及到IO的低效操作让queue_msg自己加锁完成, 不占用总的管理句柄的锁
	}
	void destoryMessageQueue(const std::string& qname)		// 销毁队列的消息管理句柄
	{
		QueueMessageManagerPtr queue_msg;
		{
			std::lock_guard<std::mutex> lock(mtx_);
			if(!existsNoLock(qname)) return;			// 队列不存在, 没得销毁
			queue_msg = queue_msgs.find(qname)->second;
			queue_msgs.erase(qname);
		}
		queue_msg->clear();
	}
	bool insert(const std::string& qname, 
				proto::message::BasicProperties* basic_basic_props,	  // 向队列新增消息
				const std::string& body, bool queue_durable)
	{
		QueueMessageManagerPtr queue_msg = getQueueMessageManagerPtr(qname);
		if(queue_msg == nullptr)
		{
			LOG_ERROR << "insert msg to the \"" << qname << "\" fail! No queue_msg_manager handle found!\n";
			return false;
		}
		return queue_msg->insert(basic_basic_props, body, queue_durable);
	}
	MessagePtr front(const std::string& qname)		// 获取队首消息
	{
		QueueMessageManagerPtr queue_msg = getQueueMessageManagerPtr(qname);
		if(queue_msg == nullptr)
		{
			LOG_ERROR << "get queue \"" << qname << "\" message fail! No queue_msg_manager handle found!\n";
			return MessagePtr();
		}
		return queue_msg->front();
	}
	bool ack(const std::string& qname, const std::string &msg_id)		 // 消息应答
	{
		QueueMessageManagerPtr queue_msg = getQueueMessageManagerPtr(qname);
		if(queue_msg == nullptr)
		{
			LOG_ERROR << "ACK queue \"" << qname << "\" message fail! No queue_msg_manager handle found!\n";
			return false;
		}
		return queue_msg->remove(msg_id);
	}	
	size_t durableCount(const std::string& qname)
	{
		QueueMessageManagerPtr queue_msg = getQueueMessageManagerPtr(qname);
		if(queue_msg == nullptr)
		{
			LOG_ERROR << "get queue \"" << qname << "\" durable_message_count fail! No queue_msg_manager handle found!\n";
			return 0;
		}
		return queue_msg->durableCount();
	}
	size_t totalCount(const std::string& qname)
	{
		QueueMessageManagerPtr queue_msg = getQueueMessageManagerPtr(qname);
		if(queue_msg == nullptr)
		{
			LOG_ERROR << "get queue \"" << qname << "\" total_message_count fail! No queue_msg_manager handle found!\n";
			return 0;
		}
		return queue_msg->totalCount();
	}
	size_t pendingPushCount(const std::string& qname)
	{
		QueueMessageManagerPtr queue_msg = getQueueMessageManagerPtr(qname);
		if(queue_msg == nullptr)
		{
			LOG_ERROR << "get queue \"" << qname << "\" pending_push_message_count fail! No queue_msg_manager handle found!\n";
			return 0;
		}
		return queue_msg->pendingPushCount();
	}
	size_t pendingAckCount(const std::string& qname)
	{
		QueueMessageManagerPtr queue_msg = getQueueMessageManagerPtr(qname);
		if(queue_msg == nullptr)
		{
			LOG_ERROR << "get queue \"" << qname << "\" pending_ACK_message_count fail! No queue_msg_manager handle found!\n";
			return 0;
		}
		return queue_msg->pendingAckCount();
	}
	void clear()
	{
		std::unique_lock<std::mutex> lock(mtx_);
		for (auto &qmsg : queue_msgs) qmsg.second->clear();
	}
private:
	QueueMessageManagerPtr getQueueMessageManagerPtr(const std::string& qname)
	{
		std::lock_guard<std::mutex> lock(mtx_);
		return existsNoLock(qname) ? queue_msgs.find(qname)->second : nullptr;
	}
	bool existsNoLock(const std::string& qname)
	{
		return queue_msgs.find(qname) != queue_msgs.end();
	}
private:
	std::mutex mtx_;
	std::string basedir_;
	std::unordered_map<std::string, QueueMessageManagerPtr> queue_msgs;		// <qname, QueueMessageManagerPtr>
};

} // namespace mqserver

#endif